package com.zyh.rcm.pvuv;

import com.zyh.rcm.entity.PageViewResult;
import com.zyh.rcm.entity.UserBehaviorEvent;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;
import java.util.concurrent.ThreadLocalRandom;

public class PVJob3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        String topicName = "topic-pv-uv";
//        SimpleStringSchema deserialization = new SimpleStringSchema();
        UserBehaviorEventDeserialization deserialization = new UserBehaviorEventDeserialization();
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka24:9092");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "flink");

        FlinkKafkaConsumer<UserBehaviorEvent> kafkaConsumer = new FlinkKafkaConsumer<>(topicName, deserialization, props);
        kafkaConsumer.assignTimestampsAndWatermarks(new UserBehaviorEventWaterMarkerAssigner(2000));

        DataStreamSource<UserBehaviorEvent> dataStream = environment.addSource(kafkaConsumer);

        //1. 先将dataStream转换为KeyedStream 为每一个元素添加一个随机数，作为其分组依据
        KeyedStream<Tuple2<Integer, UserBehaviorEvent>, Integer> keyedStream = dataStream.map(new MapFunction<UserBehaviorEvent, Tuple2<Integer, UserBehaviorEvent>>() {
            private ThreadLocalRandom random = ThreadLocalRandom.current();

            @Override
            public Tuple2<Integer, UserBehaviorEvent> map(UserBehaviorEvent value) throws Exception {
                int key = random.nextInt(16);//上限设置多少，意味以后就有多少组
                return Tuple2.of(key, value);
            }
        }).keyBy(t -> t.f0);


        //2 基于keyedStream进行窗口划分
        SingleOutputStreamOperator<PageViewResult> pageViewResultStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .aggregate(new AggregateFunction<Tuple2<Integer, UserBehaviorEvent>, Integer, Integer>() {
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }

                    @Override
                    public Integer add(Tuple2<Integer, UserBehaviorEvent> value, Integer accumulator) {
                        return accumulator + 1;
                    }

                    @Override
                    public Integer getResult(Integer accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Integer merge(Integer a, Integer b) {
                        return a + b;
                    }
                }, new ProcessWindowFunction<Integer, PageViewResult, Integer, TimeWindow>() {
                    @Override
                    public void process(Integer integer, Context context, Iterable<Integer> iterable, Collector<PageViewResult> collector) throws Exception {
                        Integer sum = iterable.iterator().next();

                        TimeWindow window = context.window();
                        long start = window.getStart();
                        long end = window.getEnd();

                        collector.collect(new PageViewResult(start, end, sum));
                    }
                });

        //pageviewResultStream包含的是各组计算后的结果  只有将各组结果汇总到一起，再累加才能获取到最终的pv
//        pageViewResultStream.print();
        SingleOutputStreamOperator<PageViewResult> result = pageViewResultStream.keyBy(pageViewResult->pageViewResult.getWindowStart()).process(new KeyedProcessFunction<Long,PageViewResult, PageViewResult>() {
            private ValueState<Long> sum;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<Long> valueStateDescriptor = new ValueStateDescriptor<>("sum", Types.LONG);
                this.sum = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(PageViewResult pageViewResult, Context context, Collector<PageViewResult> collector) throws Exception {
                Long oldSum = this.sum.value();
                if(oldSum == null){
                    oldSum = 0L;
                }

                long pv = pageViewResult.getPv();

                long newSum = pv + oldSum;
                this.sum.update(newSum);

                pageViewResult.setPv(newSum);
                //collector.collect(pageViewResult);

                //可以定义计时器:时间触发器  当水位线>=设置的时间戳之后，就会触发一个方法回调
                context.timerService().registerEventTimeTimer(pageViewResult.getWindowEnd());
            }

            //水位线>=设置的时间戳后，会自动回调onTimer方法，在这个方法中输出结果就行了
            //timestamp 设置的时间戳 就是设置的窗口关闭时间
            // ctx 工具对象
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<PageViewResult> out) throws Exception {
//                ctx.timeDomain().
                Long sum = this.sum.value();
                PageViewResult pageViewResult = new PageViewResult();
                pageViewResult.setPv(sum);
                pageViewResult.setWindowEnd(timestamp);
                Long start = ctx.getCurrentKey();
                pageViewResult.setWindowStart(start);

                out.collect(pageViewResult);


            }
        });



        result.print();

        environment.execute("pvjob");
    }
}


